package cn.edu.ccnu.nlp;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.FamilyFilter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.RegexStringComparator;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;


public class TestHbase {
	
	public static Configuration configuration;
	static {
		configuration = HBaseConfiguration.create();

	}
	
	
	

	/**
	 * 查询所有数据
	 * @param tableName
	 */
	public static void QueryAll(String tableName) {
		HTablePool pool = new HTablePool(configuration, 1000);
		HTable table = (HTable) pool.getTable(tableName);
		try {
			ResultScanner rs = table.getScanner(new Scan());
			for (Result r : rs) {
				System.out.println("获得到rowkey:" + new String(r.getRow()));
				for (KeyValue keyValue : r.raw()) {
					System.out.println("列：" + new String(keyValue.getFamily())
							+ "====值:" + new String(keyValue.getValue()));
				}
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	/**
	 * 单条件查询,根据rowkey查询唯一一条记录
	 * @param tableName
	 */
	public static void QueryByCondition1(String tableName) {

		HTablePool pool = new HTablePool(configuration, 1000);
		HTable table = (HTable) pool.getTable(tableName);
		try {
			Get scan = new Get("abcdef".getBytes());// 根据rowkey查询
			Result r = table.get(scan);
			System.out.println("获得到rowkey:" + new String(r.getRow()));
			for (KeyValue keyValue : r.raw()) {
				System.out.println("列：" + new String(keyValue.getFamily())
						+ "====值:" + new String(keyValue.getValue()));
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	/**
	 * 单条件按查询，查询多条记录
	 * @param tableName
	 */
	public static void QueryByCondition2(String tableName) {

		try {
			HTablePool pool = new HTablePool(configuration, 1000);
			HTable table = (HTable) pool.getTable(tableName);
			Filter filter = new SingleColumnValueFilter(Bytes
					.toBytes("date"), null, CompareOp.EQUAL, new RegexStringComparator(".*-.8-.*")); // 当列column1的值为aaa时进行查询
			Scan s = new Scan();
			s.setFilter(filter);
			ResultScanner rs = table.getScanner(s);
			for (Result r : rs) {
				System.out.println("获得到rowkey:" + new String(r.getRow()));
				for (KeyValue keyValue : r.raw()) {
					System.out.println("列：" + new String(keyValue.getFamily())
							+ "====值:" + new String(keyValue.getValue()));
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}

	}

	/**
	 * 组合条件查询
	 * @param tableName
	 */
	public static void QueryByYear1(String tableName) {

		try {
			
			HTablePool pool = new HTablePool(configuration, 1000);
			HTable table = (HTable) pool.getTable(tableName);

			List<Filter> filters = new ArrayList<Filter>();

			Filter filter1 = new SingleColumnValueFilter(Bytes
					.toBytes("date"), null, CompareOp.GREATER_OR_EQUAL, Bytes
					.toBytes("2006-01-01"));
			filters.add(filter1);

			Filter filter2 = new SingleColumnValueFilter(Bytes
					.toBytes("date"), null, CompareOp.LESS_OR_EQUAL, Bytes
					.toBytes("2006-12-31"));
			filters.add(filter2);

			FilterList filterList1 = new FilterList(filters);
			int sum = 0;
			String year1 = null;
			Scan scan = new Scan();
			scan.setFilter(filterList1);
			ResultScanner rs = table.getScanner(scan);
			for (Result r : rs) {
				System.out.println("获得到rowkey:" + new String(r.getRow()));
				year1 = new String(r.getRow()).substring(0, 4);
				for (KeyValue keyValue : r.raw()) {
					System.out.println("列：" + new String(keyValue.getFamily())
							+ "====值:" + new String(keyValue.getValue()));
					if((new String(keyValue.getFamily())).equals("num"))
						sum += Integer.parseInt(new String(keyValue.getValue()));
				}
			}
			rs.close();
			System.out.println("year:"+year1+"  total news:"+sum);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	
	
	public static Map QueryByYear(String tableName,String year,Map yearNum) {
		int sum = 0;
		String year_1 = null;
		try {
			HTablePool pool = new HTablePool(configuration, 1000);
			HTable table = (HTable) pool.getTable(tableName);
			Filter filter = new SingleColumnValueFilter(Bytes
					.toBytes("date"), null, CompareOp.EQUAL, new RegexStringComparator(year+"-.*-.*")); // 当列column1的值为aaa时进行查询
			Scan s = new Scan();
			s.setFilter(filter);
			ResultScanner rs = table.getScanner(s);
			for (Result r : rs) {
				year_1 = (new String(r.getRow())).substring(0,4);
				for (KeyValue keyValue : r.raw()) {
					//System.out.println("列：" + new String(keyValue.getFamily())
							//+ "====值:" + new String(keyValue.getValue()));
					if((new String(keyValue.getFamily())).equals("num"))
						sum += Integer.parseInt(new String(keyValue.getValue()));
				}
			}
			yearNum.put(year_1, sum);
			System.out.println(year_1+":"+sum);
			return yearNum;
		} catch (Exception e) {
			e.printStackTrace();
		}
		return null;
	}
	
	public Map QueryByYears(String tableName,List<String> years){
		Map<String,Integer> yearNum = new HashMap<String,Integer>();
		for(String year:years){
			yearNum = QueryByYear(tableName,year,yearNum);
		}
		
		
		System.out.println(yearNum.size());
		
		Set set = yearNum.keySet();
		Iterator it = set.iterator();
		while(it.hasNext()){
			String key = (String) it.next();
			System.out.println("key:"+key+":value"+yearNum.get(key));
		}
		return yearNum;
		
	}
	/**
	 * 单条件按查询，查询多条记录
	 * @param tableName
	 */
	public static Map QueryByMonth(String tableName,String year,String month,Map map) {

		int sum = 0;
		String month_1 = null;
		try {
			HTablePool pool = new HTablePool(configuration, 1000);
			HTable table = (HTable) pool.getTable(tableName);
			Filter filter = new SingleColumnValueFilter(Bytes
					.toBytes("date"), null, CompareOp.EQUAL, new RegexStringComparator(year+"-"+month+"-.*")); // 当列column1的值为aaa时进行查询
			Scan s = new Scan();
			s.setFilter(filter);
			ResultScanner rs = table.getScanner(s);
			for (Result r : rs) {
				//System.out.println("获得到rowkey:" + new String(r.getRow()));
				month_1 = new String(r.getRow()).substring(5,7);
				for (KeyValue keyValue : r.raw()) {
//					System.out.println("列：" + new String(keyValue.getFamily())
//							+ "====值:" + new String(keyValue.getValue()));
					if((new String(keyValue.getFamily())).equals("num"))
						sum += Integer.parseInt(new String(keyValue.getValue()));
				}
			}
			System.out.println(year+"month:"+month_1+"sum:"+sum);
			map.put(month_1, sum);
			return map;
		} catch (Exception e) {
			e.printStackTrace();
		}
		return null;
	}
	
	public Map QueryByMonths(String tableName,String year,List<String> months){
		Map<String,Integer> monthNum = new HashMap<String,Integer>();
		for(String month:months){
			monthNum = QueryByMonth(tableName,year,month,monthNum);
		}
		
		
		System.out.println(monthNum.size());
		
		Set set = monthNum.keySet();
		Iterator it = set.iterator();
		while(it.hasNext()){
			String key = (String) it.next();
			System.out.println("key:"+key+":value"+monthNum.get(key));
		}
		return monthNum;
		
	}
	/**
	 * 单条件按查询，查询多条记录
	 * @param tableName
	 */
	public static Map QueryByDay(String tableName,String year,String month,String day,Map map) {
			int sum = 0;
			String day_1 = null;
		try {
			HTablePool pool = new HTablePool(configuration, 1000);
			HTable table = (HTable) pool.getTable(tableName);
			Filter filter = new SingleColumnValueFilter(Bytes
					.toBytes("date"), null, CompareOp.EQUAL, new RegexStringComparator(year+"-"+month+"-"+day)); // 当列column1的值为aaa时进行查询
			Scan s = new Scan();
			s.setFilter(filter);
			ResultScanner rs = table.getScanner(s);
			for (Result r : rs) {
				//System.out.println("获得到rowkey:" + new String(r.getRow()));
				day_1 = new String(r.getRow()).substring(9, 10);
				for (KeyValue keyValue : r.raw()) {
					System.out.println("列：" + new String(keyValue.getFamily())
							+ "====值:" + new String(keyValue.getValue()));
					if((new String(keyValue.getFamily())).equals("num"))
						sum += Integer.parseInt(new String(keyValue.getValue()));
				}
			}
			map.put(day_1, sum);
			return map;
		} catch (Exception e) {
			e.printStackTrace();
		}
		return null;
	}
	
	public Map QueryByDays(String tableName,String year,String month,List<String> days){
		Map<String,Integer> dayNum = new HashMap<String,Integer>();
		for(String day:days){
			dayNum = QueryByDay(tableName,year,month,day,dayNum);
		}
		
		
		System.out.println(dayNum.size());
		
		Set set = dayNum.keySet();
		Iterator it = set.iterator();
		while(it.hasNext()){
			String key = (String) it.next();
			System.out.println("key:"+key+":value"+dayNum.get(key));
		}
		return dayNum;
		
	}
	
	/**
	 * get all years in hbase
	 * @param tableName
	 */
	public void getYear(String tableName){
		try {
			HTablePool pool = new HTablePool(configuration, 1000);
			HTable table = (HTable) pool.getTable(tableName);
			Filter filter = new FamilyFilter(CompareOp.GREATER_OR_EQUAL,new BinaryComparator(Bytes.toBytes("2005-01-01")));
			Scan scan = new Scan();
			scan.setFilter(filter);
			ResultScanner rs;
			rs = table.getScanner(scan);
			String year;
			List<String> years = new ArrayList<String>();
			for(Result result:rs){
				year = new String(result.getRow()).substring(0, 4);
				if(!years.contains(year)){
					years.add(year);
				}
			}
			rs.close();
			
			for(String y:years)
				System.out.println(y);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	public static void main(String[] args) throws Exception
	{
		String[] info = null;
		TestHbase th = new TestHbase();
		List<String> days = new ArrayList<String>();
		days.add("01");
		days.add("02");
		days.add("03");
		days.add("04");
//		th.createTable("blog", "base-i nfo");
//		th.insertData("news",info);
		
//		th.QueryByYears("news_stat",years);
//		th.QueryAll("news_stat");
//		th.QueryByDay("news_stat");
//		th.getYear("news_stat");
		th.QueryByDays("news_stat", "2005","08", days);
	}

}
